package com.zwh.rabbitmq.config;

import java.io.UnsupportedEncodingException;
import java.util.UUID;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;


import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zwh.rabbitmq.config.aspect.MessageProcess;
import com.zwh.rabbitmq.config.constants.RabbitTemplateConstants;
import com.zwh.rabbitmq.config.dto.RabbitmqMessageBuilder;
import com.zwh.rabbitmq.config.dto.RabbitmqMessageRecordDto;


/**
 * @author niXueChao
 * @date 2019/7/10 15:59.
 */
public class ReliableRabbitTemplate extends RabbitTemplate {
    private MessageProcess messageProcess;

    public ReliableRabbitTemplate(MessageProcess messageProcess) {
        super();
        this.messageProcess = messageProcess;
    }


    /**
     * 发送消息并保存推送消息记录
     *
     * @param businessModule 业务模块
     * @param businessKey    业务key
     * @param exchangeName   交换机名称
     * @param routingKey     路由键
     * @param data           发送数据
     */
    public void sendMsgAndSaveRecord(String businessModule, String businessKey, String exchangeName, String routingKey,
                                     Object data) {
        String sendMessageId = UUID.randomUUID().toString();
        // 消息推送记录表创建
        RabbitmqMessageRecordDto messageRecordDto = new RabbitmqMessageBuilder()
                .businessModule(businessModule)
                .businessKey(businessKey)
                .sendMessageId(sendMessageId)
                .exchangeName(exchangeName)
                .routingKey(routingKey)
                .build();
        ExtCorrelationData extCorrelationData = new ExtCorrelationData();
        extCorrelationData.setId(sendMessageId);
        extCorrelationData.setMessageInfo(messageRecordDto);
        super.convertAndSend(exchangeName, routingKey, data, extCorrelationData);
    }

    /**
     * 重写发送逻辑，新增 MessageProcess 切面触发点
     *
     * @param exchangeName
     * @param routingKey
     * @param message
     * @param correlationData
     * @throws AmqpException
     */
    @Override
    public void send(String exchangeName, String routingKey, Message message, CorrelationData correlationData) throws AmqpException {
        try {
            RabbitmqMessageRecordDto messageRecordDto = null;
            if (correlationData != null && correlationData instanceof ExtCorrelationData) {
                messageRecordDto = ((ExtCorrelationData) correlationData).getMessageInfo();
                message.getMessageProperties().setHeader(RabbitTemplateConstants.MESSAGE_HEADER_ID,
                        messageRecordDto.getSendMessageId());
                // String messageProperties = getPropertiesStr(message);
                // 设置消息发送内容
                String messageBody = getBodyStr(message);
                messageRecordDto.setContent(messageBody);
            }

            // 推送消息前调用
            messageProcess.getPreSend().accept(messageRecordDto);
            try {
                super.send(exchangeName, routingKey, message, correlationData);
            } catch (Exception e) {
                //极有可能是AmqpConnectException,长时间(和retryTemplate有关)连接不上
                // messageInfo.setFailReason(e.getMessage());
                logger.error("rabbitmq推送消息失败：" + e.getMessage());
                messageProcess.getMessageNack().accept(messageRecordDto);
            }
        } catch (Exception e) {
            //出现异常,有极有可能是保存消息出现异常
            throw new AmqpException(e.getMessage());
        }
    }


    /**
     * template 的MessageConverter 势必要 转json
     */
    String getBodyStr(Message message) {
        try {
            return new String(message.getBody(), "UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            return "";
        }
    }

    String getPropertiesStr(Message message) {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            return objectMapper.writeValueAsString(message.getMessageProperties());
        } catch (JsonProcessingException e) {
            return "";
        }
    }


    public MessageProcess getMessageProcess() {
        return messageProcess;
    }

    public void setMessageProcess(MessageProcess messageProcess) {
        this.messageProcess = messageProcess;
    }

}
